package com.shujia.core

import com.mysql.jdbc.Driver
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import java.sql.{Connection, DriverManager, Statement}

object Code21Action1 {
  def main(args: Array[String]): Unit = {
    /**
     * Action算子：
     * 并不会返回一个RDD，同时一个 Action算子 触发一个Job任务
     *
     */

    val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("CoGroup"))

    val value1RDD: RDD[(String, Int)] = sc.parallelize(List(("k1", 11), ("k1", 1), ("k2", 2), ("k3", 3), ("k5", 5), ("k4", 4), ("k3", 3), ("k3", 33)), 2)

    // 将如上数据写出至Mysql中
    value1RDD
    //      .foreach {
    //        case (key,value) => {
    //          DriverManager.registerDriver(new Driver())
    //          val connection: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/stu?useSSL=false", "root", "123456")
    //          val statement: Statement = connection.createStatement()
    //          statement.executeUpdate(
    //            /*
    //              create table spark2mysql(key varchar(20),value int)
    //             */
    //            s"""
    //              |insert into spark2mysql values('${key}',${value});
    //              |""".stripMargin)
    //          statement.close()
    //          connection.close()
    //        }
    //      }

    /**
     * foreachPartition
     *    作用：针对每个分区执行一次算子，每个分区的数据被包装成一个 迭代器，
     *        之后可以对迭代器中的数据进行遍历取值，并做计算
     *    应用场景： 针对数据写出到其他数据库中，可以调用该方法 
     */


    value1RDD
      .foreachPartition {
        case iterator => {
          DriverManager.registerDriver(new Driver())
          val connection: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/stu?useSSL=false", "root", "123456")
          val statement: Statement = connection.createStatement()
          println("Mysql链接创建成功...")
          // iterator中保存了当前分区中的数据

          // Mysql中的SQL语句通过批量方式进行插入
          //          iterator.foreach {
          //            case (key, value) => {
          //              statement.addBatch(
          //                s"""
          //                   |insert into spark2mysql values('${key}',${value});
          //                   |""".stripMargin)
          //            }
          //          }
          //          statement.executeBatch()

          var sql: String =
            s"""
               |insert into spark2mysql values
               |""".stripMargin
          iterator.foreach {
            case (key, value) => {
              sql += s"('${key}',${value}),"
            }
          }
          statement.executeUpdate(sql.reverse.tail.reverse)
          statement.close()
          connection.close()
        }
      }


  }
}
